package transport.handler;

import hdfs.TransPort;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.EventLoopGroup;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import transport.ChannelUtils;

public final class ClientHandler extends ChannelInboundHandlerAdapter {
    private static final Logger LOG = LoggerFactory.getLogger(ClientHandler.class);
    private int packetNum;
    private final EventLoopGroup aGroup;
    public ClientHandler(int packetNum, EventLoopGroup aGroup){
        this.packetNum = packetNum;
        this.aGroup = aGroup;
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ChannelUtils.tryToReadIfNeeded(ctx);
//        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
//        TransPort.WritePacketResponse response = (TransPort.WritePacketResponse) msg;
        TransPort.WritePacketRequest response = (TransPort.WritePacketRequest) msg;
        if(response.getSize()==0) {
            packetNum--;
            LOG.info("客户端{} packetTracer{}",ctx.channel().localAddress(),packetNum);
        }
        if(packetNum==0){
            ctx.close();
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        LOG.info("channel {} closed", ctx.channel());
        ChannelUtils.closeOnFlush(ctx.channel());
        aGroup.shutdownGracefully();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOG.error("data transfer handler occurs error, channel: {}", ctx.channel(), cause);
        ctx.close();
        aGroup.shutdownGracefully();
    }
}